package com.qupeng.demo.kafka.kafkaapache.security;

import org.apache.kafka.common.config.ConfigData;
import org.apache.kafka.common.config.provider.ConfigProvider;
import org.apache.kafka.common.utils.Shell;

import java.io.IOException;
import java.io.StringReader;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;

public class GpgProvider implements ConfigProvider {

    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public void close() {
    }

    @Override
    public ConfigData get(String path) {
        try {
            String passphrase = System.getenv("PASSPHRASE"); //通过环境变量PASSPHRASE来提供用于解密密码的密码短语。
            String data = Shell.execCommand( //用gpg解密配置文件。返回值中包含了全部的解密配置参数。
                    "gpg", " -- decrypt", " -- passphrase", passphrase, path);
            Properties props = new Properties();
            props.load(new StringReader(data)); //将配置数据解析成Java属性。
            Map<String, String> map = new HashMap<>();
            for (String name : props.stringPropertyNames())
                map.put(name, props.getProperty(name));
            return new ConfigData(map);
        } catch (IOException e) {
            throw new RuntimeException(e); //如果遇到错误,就抛出RuntimeException异常。
        }
    }

    @Override
    public ConfigData get(String path, Set<String> keys) { //调用者可能想要获取指定文件中的部分属性,我们在这里加载所有属性,并返回调用者请求的部分。
        ConfigData configData = get(path);
        Map<String, String> data = configData.data().entrySet()
                .stream().filter(e -> keys.contains(e.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        return new ConfigData(data, configData.ttl());
    }
}